package com.ac.scum.log.task.face.log;

import com.ac.scum.log.bean.ServerFTPFIle;
import com.ac.scum.log.task.face.log.counsumer.LogConsumer;
import com.ac.scum.log.task.face.log.counsumer.LogTypeStage;
import com.ac.scum.log.task.mapper.server.ServerFTPFileMapper;
import com.ac.scum.log.task.utils.queue.FTPQueueItem;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.LocalDateTime;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
@AllArgsConstructor
public class LogUpdateConsumer {
    private final LogTypeStage logTypeStage;
    private final StringRedisTemplate stringRedisTemplate;
    private final ServerFTPFileMapper serverFTPFileMapper;
    private RedisTemplate<String, Object> queueRedisTemplate;
    private final int processThreadMaxNum = 10;
    @SuppressWarnings("FieldMayBeFinal")
    private static ThreadPoolExecutor threadPool;

    static {
        threadPool = new ThreadPoolExecutor(
                20,
                300,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(300),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    @PreDestroy
    @SuppressWarnings("DuplicatedCode")
    public void tobeShutdown() {
        try {
            log.info("开始关闭消费者线程池");
            threadPool.shutdown();

            if (!threadPool.awaitTermination(3, TimeUnit.MINUTES)) {
                log.error("消费者线程池未能在30秒内完全关闭，执行强制关停");
                for (Runnable runnable : threadPool.shutdownNow()) {
                    log.error("消费者线程池未处理的任务：{}", runnable);
                }
            }
        } catch (InterruptedException e) {
            log.error("消费者线程池关闭过程中被打断", e);
        }
        log.info("消费者线程池已关闭");
    }

    @PostConstruct
    @Transactional
    public void serverLogsUpdateConsumer() {
        Timer timer = new Timer();
        TimerTask task = new TimerTask() {
            @Override
            public void run() {
                Long size = queueRedisTemplate.opsForList().size(LogUpdateProvider.REDIS_QUEUE_KEY);
                if ((size == null ? 0 : size) > 0) {
                    log.info("🌙当前待消费队列大小{}", size);
                    for (long i = 0L; i < Math.min(size, processThreadMaxNum); i++) {
                        threadPool.submit(() -> {
                            FTPQueueItem ftpQueueItem = (FTPQueueItem) queueRedisTemplate.opsForList().leftPop(LogUpdateProvider.REDIS_QUEUE_KEY);
                            if (ftpQueueItem == null) {
                                return;
                            }
                            try {
                                String type;
                                String cacheName = ftpQueueItem.getServer().getId() + ftpQueueItem.getName();
                                //切换临时存储数据源
//                            DynamicDataSourceContextHolder.push(ftpQueueItem.getLongStorageFlag() ? "primary" : "temp");
//                            log.info("🌙🌙 切换数据源成功 🌙🌙");
                                LogConsumer stage = logTypeStage.getStage(ftpQueueItem.getName());
                                if (ftpQueueItem.getIsNew()) {
                                    ServerFTPFIle serverFTPFIle = new ServerFTPFIle();
                                    serverFTPFIle.setServerId(ftpQueueItem.getServer().getId());
                                    serverFTPFIle.setFileName(ftpQueueItem.getName());
                                    serverFTPFIle.setFileSize(ftpQueueItem.getSize());
                                    serverFTPFIle.setType(1);
                                    serverFTPFIle.setLastUpdateTime(LocalDateTime.now());
                                    serverFTPFileMapper.insert(serverFTPFIle);
                                    ftpQueueItem.setFileId(serverFTPFIle.getId());
                                    type = "新文件已经建立";
                                } else {
                                    serverFTPFileMapper.update(Wrappers.lambdaUpdate(ServerFTPFIle.class).set(ServerFTPFIle::getFileSize, ftpQueueItem.getSize()).eq(ServerFTPFIle::getId, ftpQueueItem.getFileId()));
                                    type = "文件大小修改完毕";
                                }
                                if (stage == null) {
                                    log.error("🌙🌙 未获取到STAGE {} ", ftpQueueItem.getName());
                                    stringRedisTemplate.opsForHash().delete(LogUpdateProvider.DATA_CHECK_CACHE_KEY, cacheName);
                                } else {
                                    long start = System.currentTimeMillis();
                                    stage.consumerLog(ftpQueueItem);
                                    try {
                                        // 创建目录
                                        Path directory = Paths.get("task-log");
                                        Files.createDirectories(directory);

                                        // 写入文件
                                        Path filePath = directory.resolve(ftpQueueItem.getFileId().toString());
                                        try (OutputStreamWriter writer = new OutputStreamWriter(new FileOutputStream(filePath.toString(), true), StandardCharsets.UTF_16LE)) {
                                            writer.write(ftpQueueItem.getContext());
                                        } catch (IOException e) {
                                            log.error("写入本地文件失败", e);
                                        }
                                    } catch (IOException e) {
                                        log.error("创建目录失败", e);
                                    }
                                    stringRedisTemplate.opsForHash().delete(LogUpdateProvider.DATA_CHECK_CACHE_KEY, cacheName);
                                    log.info("🌙消费队列：服务器【{}】 文件：【{}】 {} 共计【{}s】", ftpQueueItem.getServer().getName(), ftpQueueItem.getName(), type, (System.currentTimeMillis() - start) / 1000);
                                }
                            } catch (Exception e) {
                                log.error("消费队列失败\n {}", ftpQueueItem, e);
                            }
//                        finally {
//                          DynamicDataSourceContextHolder.poll();
//                        }
                        });
                    }

                }
            }
        };
        // 每隔2秒执行一次
        timer.schedule(task, 0, 2000);


    }
}
